#include "../util/h/iUtil.h"
#include "../util/h/iLogger.h"
#include "../util/h/iThreadMgr.h"
#include "../util/h/iMsgDB.h"
#include "../util/src/iMsgThread.h"
#include "../util/src/iIOThread.h"
#include "iOTCPConnection.h"
#include "iPBNConnection.h"

#if defined(__WIN__)
#elif defined(__LINUX__)
#include <arpa/inet.h>
#endif

iTPConnection* iTPConnection::new_instance(const char* remote, const char* local, iAllocator* alloc)
{
    char rpo[32],rip[64],rpt[8];
    if(!parse_address(remote, rpo, rip, rpt))
        return NULL;

    char lpo[32],lip[64],lpt[8];
    if(!parse_address(local, lpo, lip, lpt))
        return NULL;
    
    if(strcmp(rpo, lpo))
        return NULL;

    soctx* lcs=NULL, *rcs=NULL;
    iClientSocket *cs = NULL;
    while(1)
    {
        lcs = soctx::new_instance(local);
        rcs = soctx::new_instance(remote);
        if(NULL==lcs || NULL==rcs)
            break;

        cs = new iClientSocket(lcs, rcs);
        if(!cs->bind(*lcs))
            break;
        lcs->release();rcs->release();

        if(strstr(rpo, "PKG") != NULL)
            return new(alloc->allocate(sizeof(iPBNConnection))) iPBNConnection(cs, NULL, alloc);
        else
            return new(alloc->allocate(sizeof(iOTCPConnection))) iOTCPConnection(cs, NULL, alloc);
    }

    if(lcs) lcs->release();
    if(rcs) rcs->release();
    if(cs)  delete cs;
    return NULL;
}

iOTCPConnection::iOTCPConnection(iSocket* so, iConnection* prvd, iAllocator* alloc)
:iConnection(prvd, alloc)
,m_sink(NULL)
,m_data(NULL)
{
	m_socket = so;
    m_socket->set_sink(this);
    m_jobber = (iMsgThread*)iThreadMgr::spawn_this();
    m_worker = (iMsgThread*)iThreadMgr::spawn_io(true);
}
void iOTCPConnection::release()
{
    FUNC("iOTCPConnection::release");

    if(m_conn_status == iConnection::CS_UNSPEC && m_sink==NULL)
        iConnection::release();
    else
        close();
}
// worker thread
void iOTCPConnection::on_connect_confirm()
{   
    FUNC("iOTCPConnection::on_connect_confirm");
	
    m_conn_status = CS_ESTABLISHED;
    m_jobber->post_msg(iMsg::new_instance(MSG_TP_CONNECTION_IN, NULL, this), m_worker);
}
void iOTCPConnection::on_data_indication(iMsgDB* mdb)
{
    FUNC("iOTCPConnection::on_data_indication");

    if(m_data)
        m_data->append(mdb);
    else
        m_data = mdb;

    if(m_conn_status==CS_ESTABLISHED && m_data)
    {
        m_jobber->post_msg(iMsg::new_instance(MSG_TP_DATA_IN, m_data, this), m_worker);
        m_data = NULL;
    }
}
void iOTCPConnection::on_close()
{
    FUNC("iOTCPConnection::on_close");

	m_conn_status = CS_BROKEN;
    m_socket->close();
    ((iIOThread*)m_worker)->remove_io(m_socket);
    m_jobber->post_msg(iMsg::new_instance(MSG_TP_CONNECTION_TERM, NULL, this), m_worker);
}

// jobber thread
int32_t iOTCPConnection::connect_request(iMsgDB* request, iTPConnection::iSink* sink)
{
    FUNC("iOTCPConnection::connect_request");

    if(m_conn_status == iConnection::CS_UNSPEC && m_sink==NULL)
    {
        m_sink = sink;
        m_worker->post_msg(iMsg::new_instance(MSG_TP_CONNECTION_OUT, NULL, this), m_jobber);
        return 0;
    }
	return -1;
}
int32_t iOTCPConnection::connect_response(iMsgDB* response, iTPConnection::iSink* sink)
{
    FUNC("iOTCPConnection::connect_request");

    m_sink = sink;

    return 0;
}
int32_t iOTCPConnection::close()
{
	FUNC("iOTCPConnection::close");

    m_sink = NULL;
	((iMsgThread*)m_worker)->post_msg(iMsg::new_instance(MSG_TP_CLOSE,NULL, this), (iMsgThread*)iThreadMgr::spawn_this());

	return 0;
}
int32_t iOTCPConnection::send_data(iMsgDB* data)
{
	FUNC("iOTCPConnection::send_data");

    if(m_conn_status == CS_BROKEN)
        return -1;

    if(data)
        ((iMsgThread*)m_worker)->post_msg(iMsg::new_instance(MSG_TP_DATA_OUT, data->duplicate(), this), (iMsgThread*)iThreadMgr::spawn_this());

	return 0;
}
int32_t iOTCPConnection::get_option(TP_OPTION opt, void* optval, int32_t optlen)
{
    FUNC("iOTCPConnection::get_option");

    switch(opt)
    {
    case TP_OPT_LOCAL_ADDR:
        {
            if(optlen == 64)
            {
                soctx* local = ((iClientSocket*)m_socket)->get_local();
                if(local->get_family() & AF_INET)
                    sprintf((char*)optval, "%s:%u", inet_ntoa(local->get_sa4().sin_addr), ntohs(local->get_sa4().sin_port));
            }
        }
        break;
    case TP_OPT_REMOTE_ADDR:
        {
            soctx* remote = ((iClientSocket*)m_socket)->get_remote();
            if(remote->get_family() & AF_INET)
                sprintf((char*)optval, "%s:%u", inet_ntoa(remote->get_sa4().sin_addr), ntohs(remote->get_sa4().sin_port));
        }
        break;
    case TP_OPT_PROTO_TYPE:
        {
            *((int32_t*)optval) = ORIGINAL;
        }
        break;
    default:
        return -1;
    }

    return 0;
}
int32_t iOTCPConnection::set_option(TP_OPTION opt, void* optval, int32_t optlen)
{
    FUNC("iOTCPConnection::set_option");

	return 0;
}
int iOTCPConnection::on_message(iMsgID msg, void* param)
{
    FUNC("iOTCPConnection::on_message");

    switch(msg)
    {
    case MSG_TP_DATA_IN:
        {
            if(m_sink)
                m_sink->on_data_indication((iMsgDB*)param);
            ((iMsgDB*)param)->release();
        }
        break;
    case MSG_TP_DATA_OUT:
        {
            if(m_conn_status == CS_WAIT_LOCAL_RSP)
            {
                m_conn_status = CS_ESTABLISHED;
                on_data_indication(NULL);
            }
            return iConnection::on_message(msg, param);
        }
        break;
    case MSG_TP_CONNECTION_IN:
        {
            m_sink->on_connect_confirm(0, NULL);
        }
        break;
    case MSG_TP_CONNECTION_TERM:
        {
			if(m_sink)
				m_sink->on_close(0);
            else
                iConnection::release();
        }
        break;
	case MSG_TP_CLOSE:
		{
            if(m_conn_status == CS_WAIT_LOCAL_RSP)
            {
                m_socket->close();
            }
            else if (m_conn_status == CS_BROKEN)
            {
                iConnection::release();
            }
		}
		break;
    default:
        return iConnection::on_message(msg, param);
    }

	return 0;
}
